package io.gitee.redismq.core.config;

import io.gitee.redismq.comm.constant.Constant;
import lombok.extern.slf4j.Slf4j;
import io.gitee.redismq.core.annotation.StreamListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Executors;

/**
 * @author: 孙浩
 * @since: 2025-03-05 19:22:06
 */
@Slf4j
public class RedisMqInitializer implements BeanPostProcessor {

    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    @Override
    public Object postProcessAfterInitialization(Object bean, @Nullable String beanName) {
        for (Method method : bean.getClass().getMethods()) {
            StreamListener annotation = method.getAnnotation(StreamListener.class);
            if (annotation != null) {
                String streamKey = annotation.streamKey();
                String groupName = annotation.groupName();
                String consumerName = annotation.consumerName();
                // 初始化消费者组
                createConsumerGroupIfAbsent(streamKey, groupName);
                // 解析方法参数类型
                Class<?> payloadType = resolvePayloadType(method);
                // 启动监听线程
                startListener(streamKey, groupName, consumerName, bean, method, payloadType);
            }
        }
        return bean;
    }

    private Class<?> resolvePayloadType(Method method) {
        Class<?>[] paramTypes = method.getParameterTypes();
        if (paramTypes.length != 1) {
            throw new IllegalArgumentException("The listening method must have one and only one parameter");
        }

        java.lang.reflect.Type genericType = method.getGenericParameterTypes()[0];
        return resolveGenericType(genericType);
    }

    private Class<?> resolveGenericType(java.lang.reflect.Type type) {
        if (type instanceof ParameterizedType) {
            ParameterizedType parameterizedType = (ParameterizedType) type;
            java.lang.reflect.Type[] actualTypeArgs = parameterizedType.getActualTypeArguments();
            if (actualTypeArgs.length > 0) {
                return resolveGenericType(actualTypeArgs[0]);
            }
        } else if (type instanceof Class) {
            return (Class<?>) type;
        }
        return Object.class;
    }

    private void createConsumerGroupIfAbsent(String streamKey, String groupName) {
        try {
            redisTemplate.opsForStream().createGroup(streamKey, groupName);
        } catch (RedisSystemException e) {
            if (!e.getCause().getMessage().contains("BUSYGROUP")) throw e;
        }
    }

    private void startListener(String streamKey, String groupName, String consumerName, Object bean, Method method, Class<?> payloadType) {
        Executors.newSingleThreadExecutor().submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    @SuppressWarnings("unchecked")
                    List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
                            Consumer.from(groupName, consumerName),
                            StreamReadOptions.empty().block(Duration.ofSeconds(5)),
                            StreamOffset.create(streamKey, ReadOffset.lastConsumed())
                    );
                    if (records == null || records.isEmpty()) {
                        continue;
                    }
                    for (MapRecord<String, Object, Object> record : records) {
                        Object data = record.getValue().get(Constant.DATA);
                        if (data == null) {
                            continue;
                        }
                        Object payload;
                        if (data instanceof String) {
                            //字符串类型,单独处理
                            payload = redisTemplate.getStringSerializer().deserialize(data.toString().getBytes(StandardCharsets.UTF_8));
                        } else {
                            // 直接反序列化为方法参数类型
                            payload = redisTemplate.getValueSerializer().deserialize(data.toString().getBytes(StandardCharsets.UTF_8));
                        }
                        if (payloadType.isInstance(payload)) {
                            method.invoke(bean, payloadType.cast(payload));
                            redisTemplate.opsForStream().acknowledge(streamKey, groupName, record.getId());
                        } else {
                            handleTypeMismatch(streamKey, groupName, record);
                        }
                    }
                } catch (Exception e) {
                    log.error("Error processing stream message", e);
                }
            }
        });
    }

    /**
     * 当类型不匹配时,提示即可
     */
    private void handleTypeMismatch(String streamKey, String groupName, MapRecord<String, Object, Object> record) {
        redisTemplate.opsForStream().acknowledge(streamKey, groupName, record.getId());
    }
}
